/** @file    BusMT.cpp
 *  @author  Alessandro Polo
 *  @version $Id: BusMT.cpp 2890 2010-08-12 01:57:23Z alex $
 *  @brief
 * File containing methods for the wosh::BusMT class.
 * The header for this class can be found in BusMT.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2010, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/

 #include <core/BusMT.h>
 #include <core/BusManager.h>
 #include <core/ObjectFactory.h>


using namespace std;
namespace wosh {

 WOSH_REGISTER(wosh::BusMT, "wosh::Bus", 1.01, _static_BusMT)

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

BusMT::BusMT( const std::string& busname )
	: Bus(), ThreadImpl(NULL, "wosh.Bus.BusMT") {
	if ( !busname.empty() )
		Object::setName( busname );
	 else
		 Object::setName( _BusMT_NAME, getBusID() );
	Log.setContext( Object::getName() );
	Log.setLevel( LOG_INFO );

	this->threadname = Object::getName();

	this->workersPool = new ThreadPool<BusMTWorker>();
	this->workersPool->setWorkersCount<BusMT>(2, this);
	this->workersPool->setOwnerObject(this);
	push_event( Bus::STATE_CREATED );
 }

BusMT::~BusMT() {
	Log(LOG_VERBOSE, " Destroying.." );
	if ( isBusRunning() )
		stopBus();

	forward_event_to_listeners( Bus::STATE_DESTROYED );

	// message queue free messages when destroyed

	// delete thread pool and its workers
	this->workersPool->waitApplyState(Thread::STATE_STOPPED, 20000);
	this->workersPool->setWorkersCount<BusMT>(0, this);
	delete this->workersPool; this->workersPool = NULL;

	Log(LOG_DEBUG, ":~BusRing() Destroyed." );
 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////// BUS OPERATIONS

WRESULT BusMT::startBus( unsigned long timeout ) {
	if ( isBusRunning() ) {
		Log(LOG_WARNING, ":startBus() : FAILED. Bus is already Running." );
		return WRET_ERR_ILLEGAL_USE;
	 }
	Log(LOG_INFO, ":startBus(%lu) Starting Bus..", timeout );
	WRESULT ret = startThread(timeout);
	if ( WFAILED(ret) )
		return ret;
	Log(LOG_VERBOSE, ":startBus(%lu) Starting Workers..", timeout );
	this->workersPool->waitApplyState(Thread::STATE_RUNNING, timeout);
	return ret;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT BusMT::stopBus( unsigned long timeout ) {
	if ( !isBusRunning() ) {
		Log(LOG_WARNING, ":stopBus() : FAILED. Bus is not Running." );
		return WRET_ERR_ILLEGAL_USE;
	 }
	Log(LOG_INFO, ":stopBus(%lu) : Stopping Bus..", timeout );
	WRESULT ret = quitThread(timeout);
	if ( WFAILED(ret) )
		return ret;
	Log(LOG_VERBOSE, ":stopBus(%lu) Stopping Workers..", timeout );
	this->workersPool->waitApplyState(Thread::STATE_STOPPED, timeout);
	return ret;
 }

////////////////////////////////////////////////////////////////////////////////////////////// BUS OPERATIONS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////// THREAD RUN

void BusMT::runThread() {
	forward_event_to_listeners( Bus::STATE_STARTED );
	Log(LOG_INFO, ":runThread() : Bus Started.." );

	processingInitialize();
	while( this->running )	 {
		if ( this->msgQueue.isEmpty() ) {
			ThreadImpl::sleepForMSec(75);
			setThreadAlive();
			// fix deferred (un)registration of handlers
			// as this to prevent deadlocks when connecting bus WITHIN busMessage() call
			fixDeferredListeners();
			continue;
		 }
		ThreadImpl::sleepForMSec(50);
		setThreadAlive();

		// get and remove it from queue
		Message* message = this->msgQueue.pop();
		if ( message == NULL ) continue;

		this->workersPool->getThreads().transactionBeginRead();
		BusMTWorker* bTw = this->workersPool->waitNextIdle_(10000);
		if ( bTw == NULL ) {
			Log(LOG_WARNING, " MSG %"PRId64" : Threads seems Locked [after 10sec timeout]!", message->getID() );
			bTw = this->workersPool->waitNextIdle_(-1);
		 }

		Log(LOG_INFO, "[%ld] MSG %"PRId64" %s->[%s]%s | %s", bTw->getThreadID(), message->getID(),
				message->getSource().toString().c_str(), message->getDestinationBus().c_str(), message->getDestination().toString().c_str(),
				message->getContentClassName().c_str()	);

		bTw->setMessage(message);
		this->workersPool->getThreads().transactionEnd();
	 }

	Log(LOG_INFO, ":runThread() : Bus Stopped.." );
	forward_event_to_listeners( Bus::STATE_STOPPED );
 }


////////////////////////////////////////////////////////////////////////////////////////////////// THREAD RUN
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////// MESSAGING

void BusMT::forward_message_to_listeners( const Message* message, BusDebugInfo& cDebug ) {
	if ( cDebug.isEnabled() ) {
		cDebug.setCurrentListener(NULL);
		cDebug.setCurrentMessage(message);
	 }
	if ( message == NULL ) return;
	this->nodes.transactionBeginRead();
	tIBusListenerList::Iterator it;
	const tIBusListenerList::ConstIterator it_end = this->nodes.end();
	for ( it=this->nodes.begin(); it!=it_end; ++it ) {
		if ( *it == NULL ) continue;
		if ( cDebug.isEnabled() )
			cDebug.setCurrentListener(*it);
		if ( *it == NULL ) continue;
		(*it)->busMessage( *message, this );
	 }
	this->nodes.transactionEnd();
	if ( cDebug.isEnabled() ) {
		cDebug.setCurrentListener(NULL);
		cDebug.setCurrentMessage(NULL);
	 }
	// fix deferred (un)registration of handlers
	// as this to prevent deadlocks when connecting bus WITHIN busMessage() call
	fixDeferredListeners();
 }

void BusMT::forward_event_to_listeners( Bus::BUS_STATE state ) {
	this->nodes.transactionBeginRead();
	tIBusListenerList::Iterator it;
	const tIBusListenerList::ConstIterator it_end = this->nodes.end();
	for ( it=this->nodes.begin(); it!=it_end; ++it ) {
		if ( *it == NULL ) continue;
		(*it)->busEvent( state, this );
	 }
	this->nodes.transactionEnd();
 }

/////////////////////////////////////////////////////////////////////////////////////////////////// MESSAGING
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void BusMT::setParentObject( const Object* parent ) {
	Object::setParentObject(parent);
	Thread::setOwnerObject(parent);
 }

WRESULT BusMT::setWorkersCount( int count ) {
	WRESULT ret = WRET_OK;
	this->workersPool->setWorkersCount<BusMT>(count, this);
	this->workersPool->setOwnerObject(this);
	return ret;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////// STATS

void BusMT::processingStarted() {
	MutexLockerWrite mL(&this->statsMux);
	int busyCount = this->workersPool->getThreadBusyCount();
	if ( busyCount != 0 )
		return; // then it was already started
	
	// previous processing time is saved in the Timer
	double processingTime = this->dutyCycleTimer.getTimeSpanMilliSec();
	// move the STOP timestampo to now, calculating total period
	this->dutyCycleTimer.stop();
	double period = this->dutyCycleTimer.getTimeSpanMilliSec();
	if ( period != 0 ) {
		this->dutyCycleLast = processingTime / period;
		this->dutyCycleAverage = (this->dutyCycleCount * this->dutyCycleAverage) + this->dutyCycleLast;
		this->dutyCycleAverage /= ++this->dutyCycleCount;
	}
	this->dutyCycleTimer.start();
 }

void BusMT::processingFinished() {
	int busyCount = this->workersPool->getThreadBusyCount();
	if ( busyCount != 0 )
		return; // someone is still processing
	MutexLockerWrite mL(&this->statsMux);
	this->dutyCycleTimer.stop();
 }

/////////////////////////////////////////////////////////////////////////////////////////////////////// STATS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

BusMTWorker::BusMTWorker( BusMT* parent_ptr )
	: ThreadMethod(parent_ptr, "BusMTWorker") {
	this->message = NULL;
	this->parent = parent_ptr;
	this->ownerObj = static_cast<Object*>(parent_ptr);
 }

void BusMTWorker::setMessage( Message* message_ptr ) {
	MutexLockerWrite mL(&this->processingLock);
	this->message = message_ptr;
 }

void BusMTWorker::crashDetected( double probability ) {
	this->parent->Log(LOG_CRITICAL, ":crashDetected(%f) : MsgID#%"PRId64", Listener: %s", probability, this->cDebug.getMessageId(), this->cDebug.getCurrentListener().c_str() );
 }

void BusMTWorker::work() {
	this->processingLock.lockForRead();
	if ( this->message == NULL ) {
		this->processingLock.unLock();
		ThreadImpl::sleepForMSec(50);
		setThreadAlive();
		return;
	 }
	setThreadAlive();
	this->parent->processingStarted();
	this->processing = true;
	// forward message to handlers (message is passed as constant)
	if ( this->parent->isForMyself(message) )
		this->parent->forward_message_to_listeners( this->message, this->cDebug );
	setThreadAlive();
	// forward message to the Bus Manager, it will route message (if required)
	this->parent->push_message( this->message );
	// finally delete the instance
	delete this->message; this->message = NULL;
	this->processing = false;
	this->processingLock.unLock();
	this->parent->processingFinished();
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

}; // namespace wosh
